var zookeeper = require('node-zookeeper-client');
var amqp = require('amqplib/callback_api');
var util = require('util');
var EventEmitter = require('events').EventEmitter;

// zk
var __zkNode = '/myrpc/server/currid';
var __zkConfig = 'localhost:2181';
var __zkBaseId = '1';
// mq
var __rabbitmqConfig = 'amqp://localhost';
var __qRequest = 'request_to_{type}';
var __qReply = 'reply_to_{appId}';


// all events
// error, err
// zkDone
// mqDone
function Wrapper() {}

util.inherits(Wrapper, EventEmitter);


Wrapper.prototype.init = function(type, handler, cb) {
	console.log('init');
	this.type = type;
	this.handler = handler;
	this.initCb = cb;
	this.prepareZk();
};

Wrapper.prototype.prepareZk = function() {
	console.log('prepareZk');
	var self = this;
	var zk = zookeeper.createClient(__zkConfig);
	self.__zk = zk;
	zk.connect();

	function setMyId(id) {
		zk.setData(__zkNode, new Buffer(id + ''), -1, function(err, stat) {
			if (err) return self.emit('error', err);
			self.appId = id;
			self.emit('zkDone');
		});
	}

	function getId() {
		zk.getData(__zkNode, null, function(err, data, stat) {
			console.log('getData node %s', __zkNode);
			if (err) return self.emit('error', err);
			var myId = Number(data) + 1;
			self.appId = myId;
			setMyId(myId);
		});
	}

	function firstSetId() {
		zk.mkdirp(__zkNode, new Buffer(__zkBaseId), function(err, path) {
			console.log('create node %s', __zkNode);
			if (err) return self.emit('error', err);
			self.appId = __zkBaseId;
			self.emit('zkDone');
		});
	}


	zk.once('connected', function() {
		zk.exists(__zkNode, function(err, stat) {
			if (err) return self.emit('error', err);
			if (stat) {
				console.log('exists node %s', __zkNode);
				getId();
			} else {
				console.log('not exists node %s', __zkNode);
				firstSetId();
			}
		});
	});
};

Wrapper.prototype.prepareMq = function() {
	console.log('prepareMq');
	var self = this;
	amqp.connect(__rabbitmqConfig, function(err, conn) {
		if (err) return self.emit('error', err);

		conn.createChannel(function(err, ch) {
			if (err) return self.emit('error', err);
			self.__mqCh = ch;
			//
			prepareMqReplyChannel();
		});
	});

	function prepareMqReplyChannel() {
		ch = self.__mqCh;
		var qReply = __qReply.replace('{appId}', self.appId);
		ch.assertQueue(qReply, {
			exclusive: true
		}, function(err, q) {
			console.log('create queue %s', qReply);
			ch.consume(qReply, function(msg) {
				ch.ack(msg); // ack
				console.log('receive reply: %s', msg.content);
				var msgId = msg.properties.correlationId;
				if (!self.rpcCallback) {
					self.rpcCallback = {};
				}
				var cb = self.rpcCallback[msgId];
				if (!cb) {
					console.error('no callback with msgId = %s', msgId);
				} else {
					cb(msg.content);
				}
			});
			prepareMqRequestChannel();
		});
	}

	function prepareMqRequestChannel() {
		var ch = self.__mqCh;
		var qRequst = __qRequest.replace('{type}', self.type);
		console.log('create queue %s', qRequst);
		ch.assertQueue(qRequst, {
			durable: true
		});
		ch.prefetch(1);
		ch.consume(qRequst, function(msg) {
			console.log('receive request:%s', msg.content);
			ch.ack(msg);
			self.handler(msg, function(result) {
				var replyTo = msg.properties.replyTo;
				var replyQid = __qReply.replace('{appId}', replyTo);
				console.log('handle done, replyQid=%s, result=%s', replyQid, result);
				ch.sendToQueue(replyQid, new Buffer(result), {
					correlationId: msg.properties.correlationId
				});
			});
		}, {
			noAck: false
		});
		self.emit('mqDone');
	}
}


Wrapper.prototype.rpc = function(type, arg, cb) {
	console.log('rpc', type, arg);
	var self = this;
	var ch = self.__mqCh;
	// send to mq
	var qSend = __qRequest.replace('{type}', type);
	ch.assertQueue(qSend, {
		durable: true
	}, function(err, q) {
		var msgId = self.nextMsgId() + '';
		ch.sendToQueue(qSend, new Buffer(arg), {
			correlationId: msgId,
			replyTo: self.appId + '',
			persisitent: true,
		});

		// put callback to list
		if (!self.rpcCallback) {
			self.rpcCallback = {};
		}
		self.rpcCallback[msgId] = cb;
	});
}

Wrapper.prototype.nextMsgId = function() {
	var self = this;
	if (!self.msgId) self.msgId = 1;
	else self.msgId++;
	return self.msgId + '';
}



var wrapper = new Wrapper();
wrapper.on('zkDone', function() {
	this.prepareMq();
});

wrapper.on('mqDone', function() {
	this.initCb(null, this);
});

wrapper.on('error', function(err) {
	this.initCb(err);
})


module.exports = wrapper;